﻿#include"commConnectionPool.h"
#include"public.h"

//线程安全的懒汉单例模式接口
ConnectionPool* ConnectionPool::getConnectionPool() {
	//静态局部变量由编译器自动lock和unlock
	static ConnectionPool pool;
	return &pool;
}

//从线程池中获取线程
shared_ptr<Connection> ConnectionPool::getConnection() {
	unique_lock<mutex> lock(_queueMutex);
	while (_connectionQue.empty()) {
		//wait_for阻塞函数，直到线程通知调用或者超时则结束
		if (cv_status::timeout == cv.wait_for(lock, chrono::milliseconds(_connectionTimeout)))
		{
			if (_connectionQue.empty()) {
				LOG("获取空闲连接超时，获取失败!");
				return nullptr;
			}
		}
	}
	/*
	shared_ptr智能指针析构时，会把connection资源直接给delete掉，
	相当于调用了Connection的析构函数，connect就会被关闭掉了，
	这里需要自定义shared_ptr的释放资源方式，把connection归还到队列中
	*/
	shared_ptr<Connection> sp(_connectionQue.front(),
		[&](Connection* pcon) {
			//这里是在服务器应用线程中调用的，所以一定要考虑线程安全
			unique_lock<mutex> lock(_queueMutex);
			//智能指针析构的时候会将指针重新输入到队列中
			_connectionQue.push(pcon);
			pcon->refreshAliveTime();
		});
	_connectionQue.pop();
	cv.notify_all();//消费完连接之后，通知生产者线程检查一下，如果生产队列为空后，就通知线程赶紧生产
	return sp;
}

ConnectionPool::ConnectionPool() {
	if (!loadConfigFile()) {
		return;
	}
	for (int i = 0; i < _initSize; i++) {
		Connection* p = new Connection();
		p->connect(_ip, _port, _username, _password, _dbname);
		p->refreshAliveTime();
		_connectionQue.push(p);
		_connectionCnt++;
	}
	//启动一个新的线程，作为一个连接的生产者
	thread produce(std::bind(& ConnectionPool::produceConnectionTask, this));
	//分离线程（分离线程），主线程结束后该线程自动结束
	produce.detach();
	//启动一个定时线程，扫描超过maxIdleTime时间的空闲线程
	thread scanner(std::bind(&ConnectionPool::scannerConnectionTask, this));
	scanner.detach();
} 


//用来扫描连接的进程函数，防止很线程池中线程数量大于initSize而不归还资源
void ConnectionPool::scannerConnectionTask()
{
	while (true) 
	{
		//通过sleep模拟定时效果
		this_thread::sleep_for(chrono::seconds(_maxIdleTime));

		//扫秒整个队列，释放多余的连接
		//队列要用互斥锁，防止多线程访问
		unique_lock<mutex> lock(_queueMutex);
		while (_connectionCnt > _initSize)
		{
			Connection* p = _connectionQue.front();
			//如果队列头都没有超时的话，那么后面的connection肯定不会超时
			//每次回收返回队列都是插入在队尾的
			if (p->getAliveTime() >= _maxIdleTime)
			{
				_connectionQue.pop();
				_connectionCnt--;
				delete p;
			}
			else {
				break;
			}
		}
	}
}

//用来线程来产生连接
void ConnectionPool::produceConnectionTask() {
	while (true) {
		//所有的线程在创建时都被_queueMutex锁住了，共用一把锁，函数作用域结束后默认解锁
		//unique_lock无默认参数时会自动加锁
		unique_lock<mutex> lock(_queueMutex);
		while (!_connectionQue.empty()) {
			//condition_variable cv 必须和unique_lock一起使用
			cv.wait(lock);//队列不为空，此处生产者线程进入等待状态
		}
		if (_connectionCnt < _maxSize) {
			Connection* p = new Connection();
			p->connect(_ip, _port, _username, _password, _dbname);
			p->refreshAliveTime();
			_connectionQue.push(p);
			_connectionCnt++;
		}
		cv.notify_all();//通知消费者线程可以进行连接了
	}
}


bool ConnectionPool::loadConfigFile() {
	//读取配置文件
	FILE* pf = fopen("./mysql.ini", "r");
	if (pf == nullptr) {
		LOG("mysql.ini file is not exist!");
		return false;
	}
	//feof()函数判断文件字节流是否到了末尾
	while (!feof(pf)) {
		char line[1024] = { 0 };
		//fgets获取文件一行，并指定行的最大值（包含最后的空字符）
		//如果成功，该函数返回相同的 str 参数。
		//如果到达文件末尾或者没有读取到任何字符，str 的内容保持不变，并返回一个空指针。
		//如果发生错误，返回一个空指针。
		fgets(line, 1024, pf);
		string str = line;
		int idx=str.find('=', 0);
		if (idx == -1) {
			//无效的配置
			continue;
		}
		int endx = str.find('\n', idx);
		string key = str.substr(0, idx);
		string value = str.substr(idx + 1, endx - idx - 1);

		if (key == "ip") {
			_ip = value;
		}
		else if (key == "port") {
			_port = atoi(value.c_str());
		}
		else if (key == "username") {
			_username = value;
		}
		else if (key == "password") {
			_password = value;
		}
		else if (key == "dbname") {
			_dbname = value;
		}
		else if (key == "initSize") {
			_initSize = atoi(value.c_str());
		}
		else if (key == "maxSize") {
			_maxSize = atoi(value.c_str());
		}
		else if (key == "maxIdleTime") {
			_maxIdleTime = atoi(value.c_str());
		}
		else if (key == "connectionTimeout") {
			_connectionTimeout = atoi(value.c_str());
		}
	}
	return true;
}
